package cn.achuan.spring.springcloud.demo;

import cn.achuan.spring.springcloud.demo.product.Product;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;

/**
 * 商品消息监听器
 */
@EnableBinding(Sink.class)
public class ProductMsgListener {
    private static final Logger logger = LoggerFactory.getLogger(ProductMsgListener.class);

    @Autowired
    private ProductService productService;

    /**
     * Sink是Spring Cloud Stream提供的一个抽象消息监听接口，当在@StreamListener注解中添加了
     * 该接口类名之后，Stream就会向Kafka添加一个消息订阅，所订阅的消息主题就是我们在配置文件指定
     * 的product-topic，当主题中有消息时，Stream就会将该主题中的消息反序列化为ProductMsg，然
     * 后执行具体的消息监听方法。
     */
    @StreamListener(Sink.INPUT)
    public void onProductMsg(ProductMsg productMsg) {
        if (ProductMsg.P_UPDATE.equalsIgnoreCase(productMsg.getAction())) {
            logger.debug("收到商品变更消息，商品货号: {}", productMsg.getItemCode());
            // 重新获取该商品信息
            Product product = this.productService.loadByItemCode(productMsg.getItemCode());
            if (null != product)
                logger.debug("重新获取到的商品信息为:{}", product);
            else
                logger.debug("货号为:{} 的商品不存在", productMsg.getItemCode());
        } else if (ProductMsg.P_DELETE.equalsIgnoreCase(productMsg.getAction())) {
            logger.debug("收到商品删除消息，所要删除商品货号为: {}", productMsg.getItemCode());
        } else {
            logger.debug("收到未知商品消息: {}", productMsg);
        }
    }
}